package syan.redisp.controller;

import syan.redisp.service.RedisService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.Map;

/**
 * @author CHENTAIWEN
 * @create 2022/8/1 15:51
 */
@RestController
public class RedisPS {

    @Autowired
    private RedisTemplate<String, String> redisTemplate;
    @Autowired
    private RedisService redisService;

    @PostMapping("/query")
    public Object query(String channel){
//        return redisService.query(channel);
//        XRANGE key start end [COUNT count]

        return redisTemplate.opsForStream().read(StreamOffset.fromStart(key+channel));
    }

    @PostMapping("/add")
    public void add(String channel){
//        redisService.add(channel);
        Map<String, Object> map = new HashMap<>();
        String mapId = System.currentTimeMillis()+"-0";
        map.put(mapId,"下单");
        redisTemplate.opsForStream().add(key+channel,map);
    }

    @PostMapping("/update")
    public void update(String channel, String body){
        Map<String, Object> map = new HashMap<>();
        String mapId = System.currentTimeMillis()+"-0";
        map.put(mapId,"下单");
        redisTemplate.opsForStream().add(key+channel,map);
    }

    @PostMapping("/delete")
    public void delete(String channel){
        redisService.delete(channel);
    }

    //    Publish 发布 subscription 订阅
    private static String key="Message:Order:";

    @PostMapping("/queryConsumer")
    public Object queryConsumer(String channel, String consumer, String value){
//XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
        return redisTemplate.opsForStream().acknowledge(key+channel,consumer, value);
    }

    @PostMapping("/readConsumer")
    public Object readConsumer(String channel,String name,String group){
//XREADGROUP GROUP consumer-group-name consumer-name COUNT 1 STREAMS mystream >
        Consumer consumer=Consumer.from(group,name);
//        return redisTemplate.boundStreamOps(key+channel).read(consumer,ReadOffset.from(">"));
        return redisTemplate.opsForStream().read(consumer,StreamOffset.create(key+channel, ReadOffset.from(">")));
    }

    @PostMapping("/addConsumer")
    public Object addConsumer(String channel,String consumer){
//        XGROUP CREATE mystream consumer-group-name 0-0 0-0 从首部 $尾部开始消费，只接受新消息，当前 Stream 消息会全部忽略。
//        createGroup默认是尾部需要处理
        return redisTemplate.opsForStream().createGroup(key+channel, ReadOffset.from("0-0"),consumer);
    }
}